package com.winning.cjfw.service.service;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.winning.cjfw.common.utils.CollectionUtil;
import com.winning.cjfw.service.commons.Constant;
import com.winning.cjfw.service.commons.JobExcuteReuest;
import com.winning.cjfw.service.commons.TableInfo;
import com.winning.cjfw.service.entity.dto.ClientRecordCount;
import com.winning.cjfw.service.entity.po.ClientRecord;
import com.winning.cjfw.service.jdbc.BatchInsertUtils;
import com.winning.cjfw.service.jdbc.SpringContextHolder;
import com.winning.cjfw.service.mapper.DataBaseMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Author: yanxf@winning.com.cn
 * Description: 采集服务业务类
 * Date 2019/05/25 9:48
 */
@Service
@Slf4j
public class CollectService {

    @Resource
    private BatchInsertUtils batchInsertUtils;

    @Resource
    private DataBaseMapper dataBaseMapper;

    @Resource
    private ClientRecordService clientRecordService;


    /**
     * 查询数据量大小
     */
    private static final int SIZE = 100000;

    /**
     * 采集入口
     *
     * @param jobExcuteReuest 作业执行请求实体
     */
    public void collectTables(JobExcuteReuest jobExcuteReuest) {
        String id = jobExcuteReuest.getId();
        String nodeid = jobExcuteReuest.getNodeid();

        List<TableInfo> tablelist = jobExcuteReuest.getTablelist();


        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( 3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new ThreadFactoryBuilder().setNameFormat( "collectTables-pool-%d" ).build() );


        for (TableInfo tableInfo : tablelist) {
            threadPoolExecutor.execute( () -> collectSingleTable( id, nodeid, tableInfo ) );
        }

        threadPoolExecutor.shutdown();
    }


    /**
     * 单表采集
     */
    private void collectSingleTable(String id, String nodeid, TableInfo tableInfo) {
        //1、质控数据统计
        sumByDate( tableInfo );


        String tablename = tableInfo.getTablename();

        //同时支持sqlserver、mysql
        StringBuilder querySql = new StringBuilder();
        //sqlserver
        querySql.append( "select top (" ).append( SIZE ).append( ") * FROM " ).append( tablename );
        //mysql
//      querySql.append( "select * FROM " ).append( tablename ).append( " limit " ).append( SIZE );
        JdbcTemplate jdbcTemplate = SpringContextHolder.getBean( JdbcTemplate.class );
        List<Map<String, Object>> datas = jdbcTemplate.queryForList( querySql.toString() );

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( 3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new ThreadFactoryBuilder().setNameFormat( "collectSingleTable-pool-%d" ).build() );

        List<List<Map<String, Object>>> lists = CollectionUtil.splitList( datas, 2000 );
        List<String> keyCols = getKeyCols( tablename );
        lists.forEach( list -> threadPoolExecutor.execute( () -> collectFlow( tablename, keyCols, list, jdbcTemplate ) ) );
        threadPoolExecutor.shutdown();
    }

    /**
     * 质控数据统计
     * @param tableInfo 表名、日期列对应实体
     */
    private void sumByDate(TableInfo tableInfo) {
        String tablename = tableInfo.getDatecolumn();
        String datecolumn = tableInfo.getDatecolumn();

        List<ClientRecordCount> counts = dataBaseMapper.countByDate( tablename, datecolumn );

        clientRecordService.delete( counts, tablename );

        counts.forEach( clientRecordCount -> save( clientRecordCount, tablename ) );

    }

    /**
     * 保存质控汇总记录
     */
    private void save(ClientRecordCount clientRecordCount, String tablename) {
        ClientRecord clientRecord = new ClientRecord();
        clientRecord.setJycwsl( 0 );
        clientRecord.setJyqsl( 0 );
        clientRecord.setJyzqsl( 0 );
        clientRecord.setJysj( new Date() );
        clientRecord.setTbrq( new Date() );
        clientRecord.setSczt( Constant.SCZT_SUCC );
        clientRecord.setLsh( tablename + clientRecordCount.getDate() );
        clientRecord.setYljgdm( clientRecordCount.getYljgdm() );
        clientRecord.setYwxtscsl( clientRecordCount.getCount() );
        clientRecord.setYwxtycsl( clientRecordCount.getCount() );
        clientRecordService.insert( clientRecord );
    }

    /**
     * 采集流程
     *
     * @param tablename 表名
     * @param keyCols   主键列
     * @param list      处理数据
     */
    private void collectFlow(String tablename, List<String> keyCols, List<Map<String, Object>> list, JdbcTemplate jdbcTemplate) {
        //2、kafka消息封装、投递
        //3、备份到历史log表
        String logTablename = tablename + "_LOG";
        try {
            batchInsertUtils.execute( list, logTablename );
        } catch (Exception e) {
            log.error( logTablename + "历史数据备份异常：", e );
            //若失败根据主键将原表上传状态字段改为2，表示上传异常。更新日志记录表当前消息和调度任务处理异常
        }

        //4、根据主键删除原表数据
        StringBuilder delSql = new StringBuilder();
        delSql.append( "delete from " ).append( tablename ).append( " where " );
        keyCols.forEach( colname -> delSql.append( colname ).append( "=? and " ) );
        String delSqlStr = StringUtils.substring( delSql.toString(), 0, delSql.lastIndexOf( "and" ) );
        List<Object[]> collects = list.stream().map( data -> {
            Object[] obj = new Object[keyCols.size()];
            for (int i = 0; i < keyCols.size(); i++) {
                obj[i] = data.get( keyCols.get( i ) );
            }
            return obj;
        } ).collect( Collectors.toList() );


        jdbcTemplate.batchUpdate( delSqlStr, collects );
    }


    /**
     * 获取主键列名
     *
     * @param tablename 表名
     * @return 主键列
     */
    private List<String> getKeyCols(String tablename) {
        JdbcTemplate jdbcTemplate = SpringContextHolder.getBean( JdbcTemplate.class );
        List<Map<String, Object>> list = jdbcTemplate.queryForList( "SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE WHERE TABLE_NAME= '" + tablename + "'" );

        List<String> colNames = Lists.newArrayList();
        list.forEach( map -> {
            ArrayList values = new ArrayList( map.values() );
            colNames.addAll( values );
        } );
        return colNames;
    }


}
